#define LOG_NDEBUG 0
#define LOG_TAG "[native] RTSP"

#include <android/log.h>
#include <pthread.h>
#include <signal.h>
#include <fcntl.h>
#include <jni.h>
#include "BasicUsageEnvironment.hh"
#include "H264VideoRTPSource.hh"
#include "OutputFile.hh"
#include "liveMedia.hh"
#include "RTMPSink.h"
#include "debug.h"

/* for logging */
#define ALOGLT(level, tag, fmt, args...) __android_log_print((level), (tag), (fmt), ##args)
#ifdef ALOGD
#undef ALOGD
#endif
#ifdef ALOGE
#undef ALOGE
#endif
#define ALOGD(fmt, args...) ALOGLT(ANDROID_LOG_INFO, LOG_TAG, fmt, ##args)
#define ALOGE(fmt, args...) ALOGLT(ANDROID_LOG_ERROR, LOG_TAG, fmt, ##args)
#define ALOGDT(tag, fmt, args...) ALOGLT(ANDROID_LOG_INFO, tag, fmt, ##args)
#define ALOGET(tag, fmt, args...) ALOGLT(ANDROID_LOG_ERROR, tag, fmt, ##args)
#define ALOGDV(fmt, args...) ALOGDT(LOG_TAG " VIDEO", fmt, ##args)
#define ALOGEV(fmt, args...) ALOGET(LOG_TAG " VIDEO", fmt, ##args)
#define ALOGDA(fmt, args...) ALOGDT(LOG_TAG " AUDIO", fmt, ##args)
#define ALOGEA(fmt, args...) ALOGET(LOG_TAG " AUDIO", fmt, ##args)

// Forward function definitions:
// RTSP 'response handlers':
static void continueAfterDESCRIBE(RTSPClient* rtspClient, int resultCode,
        char* resultString);
static void continueAfterSETUP(RTSPClient* rtspClient, int resultCode,
        char* resultString);
static void continueAfterPLAY(RTSPClient* rtspClient, int resultCode,
        char* resultString);

// Other event handler functions:
static void subsessionAfterPlaying(void* clientData); // called when a stream's subsession (e.g., audio or video substream) ends
static void subsessionByeHandler(void* clientData); // called when a RTCP "BYE" is received for a subsession
// called at the end of a stream's expected duration (if the stream has not already signaled its end using a RTCP "BYE")
static void streamTimerHandler(void* clientData);

// Used to iterate through each stream's 'subsessions', setting up each one:
static void setupNextSubsession(RTSPClient* rtspClient);
// Used to shut down and close a stream (including its "RTSPClient" object):
// void shutdownStream(RTSPClient* rtspClient, int exitCode = 1);
static void shutdownStream(RTSPClient* client, int exitCode = 1);

// A function that outputs a string that identifies each stream (for debugging output).  Modify this if you wish:
static UsageEnvironment& operator<<(UsageEnvironment& env,
        const RTSPClient& rtspClient) {
    return env << "[URL:\"" << rtspClient.url() << "\"]: ";
}

// A function that outputs a string that identifies each subsession (for debugging output).  Modify this if you wish:
static UsageEnvironment& operator<<(UsageEnvironment& env,
        const MediaSubsession& subsession) {
    return env << subsession.mediumName() << "/" << subsession.codecName();
}

typedef struct {
    const char *mUrl;
    bool mRtcpSendBye;
    char mQuit;
    UsageEnvironment* mEnv;
    MediaSession* mSession;
    RTSPClient* mClient;
    pthread_t mThreadId;
    jmethodID mVideoEvent;
    jmethodID mAudioEvent;
    jmethodID mExitEvent;
    void *mRTMPSink;
    pthread_mutex_t mRTMPSinkLock;
    bool mStarted;
} CustomClient;

static pthread_mutex_t gRTMPSinkLock = PTHREAD_MUTEX_INITIALIZER;
#define LOCK(l) pthread_mutex_lock(&(l))
#define UNLOCK(l) pthread_mutex_unlock(&(l))
static void *gRTMPSink = NULL;
#define RTMP_VIDEO_CODEC CODECTYPE_VIDEO_H264
#define RTMP_AUDIO_CODEC CODECTYPE_AUDIO_AAC
static volatile int gRTMPChannel = 0;

class Autolock {
public:
        Autolock(pthread_mutex_t *lock) : mLock(lock) {
            if (mLock) {
                pthread_mutex_lock(mLock);
            }
        }

        ~Autolock() {
            if (mLock) {
                pthread_mutex_unlock(mLock);
            }
        }
private:
    pthread_mutex_t *mLock;
};

// Define a class to hold per-stream state that we maintain throughout each stream's lifetime:
class StreamClientState {
public:
    StreamClientState();
    virtual ~StreamClientState();

public:
    MediaSubsessionIterator* iter;
    MediaSession* session;
    MediaSubsession* subsession;
    TaskToken streamTimerTask;
    double duration;
};

class ourRTSPClient: public RTSPClient {
public:
    static ourRTSPClient* createNew(UsageEnvironment& env, void * client_data,
            int verbosityLevel = 0, char const* applicationName = NULL,
            portNumBits tunnelOverHTTPPortNum = 0);

protected:
    ourRTSPClient(UsageEnvironment& env, void * client_data, int verbosityLevel,
            char const* applicationName, portNumBits tunnelOverHTTPPortNum);
    // called only by createNew();
    virtual ~ourRTSPClient();

public:
    StreamClientState scs;
    CustomClient *mCustomClient;
};

static jclass gClass;
static jobject gObject;
static JavaVM *g_JavaVM;

static pthread_mutex_t gObjectLock = PTHREAD_MUTEX_INITIALIZER;
static void LockGlobalObject() {
    pthread_mutex_lock(&gObjectLock);
}
static void UnlockGlobalObject() {
    pthread_mutex_unlock(&gObjectLock);
}

static CustomClient gClients[3] = {
    [0] = {.mUrl = NULL, .mClient = NULL, .mRTMPSink = NULL, .mRTMPSinkLock = PTHREAD_MUTEX_INITIALIZER, .mStarted = false},
    [1] = {.mUrl = NULL, .mClient = NULL, .mRTMPSink = NULL, .mRTMPSinkLock = PTHREAD_MUTEX_INITIALIZER, .mStarted = false},
    [2] = {.mUrl = NULL, .mClient = NULL, .mRTMPSink = NULL, .mRTMPSinkLock = PTHREAD_MUTEX_INITIALIZER, .mStarted = false},
};

struct fields_t {
    jmethodID post_video_channel0;
    jmethodID post_audio_channel0;
    jmethodID post_exit_channel0;
    jmethodID post_video_channel1;
    jmethodID post_audio_channel1;
    jmethodID post_exit_channel1;
    jmethodID post_video_channel2;
    jmethodID post_audio_channel2;
    jmethodID post_exit_channel2;
};

static void* SetupClient(void* rv);
static void init();
static JNIEnv *GetEnv();
static void DetachCurrent();
static void videonotify(CustomClient * client, int msg, const jbyteArray cbb,
        long presentationTime);
static void audionotify(CustomClient * client, int msg, const jbyteArray cbb,
        long presentationTime);
static void exitnotify(CustomClient *client);

//#define FILE_SAVE_ENB
//#define AUDIO_FILE_SAVE_ENB
#define CHECK_IDRFRM_ENB
#define PREPEND_SPSPPS_TO_IDRFRM
class HEVCSink: public MediaSink {
public:
    static HEVCSink* createNew(UsageEnvironment& env,
            MediaSubsession& subsession, // identifies the kind of data that's being received
            void * client_data); // identifies the stream itself (optional)
private:
    HEVCSink(UsageEnvironment& env, MediaSubsession& subsession,
            void * client_data);
    // called only by "createNew()"
    virtual ~HEVCSink();
    static void afterGettingFrame(void* clientData, unsigned frameSize,
            unsigned numTruncatedBytes, struct timeval presentationTime,
            unsigned durationInMicroseconds);
    virtual void afterGettingFrame(unsigned frameSize,
            unsigned numTruncatedBytes, struct timeval presentationTime);
    void addData(unsigned char const* data, unsigned dataSize);

private:
    // redefined virtual functions:
    RTSPClient * mClient;
    virtual Boolean continuePlaying();
    char const* fSPropParameterSetsStr[3];
    Boolean fHaveWrittenFirstFrame;
    unsigned char* fBuffer;
    struct timeval fPrevPresentationTime;
#ifdef FILE_SAVE_ENB
    FILE* fOutFid;
#endif /* FILE_SAVE_ENB */
#ifdef AUDIO_FILE_SAVE_ENB
    FILE* fAudioOutFid;
#endif /* AUDIO_FILE_SAVE_ENB */
private:
	MediaSubsession& fSubsession;
#ifdef CHECK_IDRFRM_ENB
	char codecName[16];
#endif /* CHECK_IDRFRM_ENB */
	char* fStreamId;
#ifdef FILE_SAVE_ENB
	int fd;
#endif /* FILE_SAVE_ENB */
#ifdef AUDIO_FILE_SAVE_ENB
	int audio_fd;
#endif /* AUDIO_FILE_SAVE_ENB */
};

#define RTSP_CLIENT_VERBOSITY_LEVEL 1 // by default, print verbose output from each "RTSPClient"
void continueAfterDESCRIBE(RTSPClient* rtspClient, int resultCode,
        char* resultString);

void sendDescribe(void *rtspClient) {
    ((RTSPClient *)rtspClient)->sendDescribeCommand(continueAfterDESCRIBE);
}

void continueAfterDESCRIBE(RTSPClient* rtspClient, int resultCode,
        char* resultString) {
    CustomClient *client = ((ourRTSPClient *) rtspClient)->mCustomClient;
    do {
        UsageEnvironment& env = rtspClient->envir(); // alias
        StreamClientState& scs = ((ourRTSPClient*) rtspClient)->scs; // alias
        if (resultCode != 0) {
            __android_log_print(ANDROID_LOG_INFO, "native",
                    "Failed to get a SDP description:%s", resultString);
            delete[] resultString;
            break;
        }
        char* const sdpDescription = resultString;
        scs.session = MediaSession::createNew(env, sdpDescription);
        client->mSession = scs.session;
        delete[] sdpDescription; // because we don't need it anymore
        if (scs.session == NULL) {
            __android_log_print(ANDROID_LOG_INFO, "native",
                    "Failed to create a MediaSession object from the SDP description:%s",
                    env.getResultMsg());
            break;
        } else if (!scs.session->hasSubsessions()) {
            __android_log_print(ANDROID_LOG_INFO, "native",
                    "This session has no media subsessions (i.e., no \"m=\" lines)\n");
            break;
        }

        scs.iter = new MediaSubsessionIterator(*scs.session);
        setupNextSubsession(rtspClient);
        return;
    } while (0);
    rtspClient->envir().taskScheduler().scheduleDelayedTask(3000000, (TaskFunc*)sendDescribe, rtspClient);
}

#define REQUEST_STREAMING_OVER_TCP False

void setupNextSubsession(RTSPClient* rtspClient) {
    UsageEnvironment& env = rtspClient->envir(); // alias
    StreamClientState& scs = ((ourRTSPClient*) rtspClient)->scs; // alias
    scs.subsession = scs.iter->next();
    if (scs.subsession != NULL) {
        if (!scs.subsession->initiate()) {
            env << *rtspClient << "Failed to initiate the \"" << *scs.subsession
                    << "\" subsession: " << env.getResultMsg() << "\n";
            setupNextSubsession(rtspClient); // give up on this subsession; go to the next one
        } else {
            env << *rtspClient << "Initiated the \"" << *scs.subsession
                    << "\" subsession (";
            if (scs.subsession->rtcpIsMuxed()) {
                env << "client port " << scs.subsession->clientPortNum();
            } else {
                env << "client ports " << scs.subsession->clientPortNum() << "-"
                        << scs.subsession->clientPortNum() + 1;
            }
            env << ")\n";

            rtspClient->sendSetupCommand(*scs.subsession, continueAfterSETUP,
                    False, REQUEST_STREAMING_OVER_TCP);
        }
        return;
    }

    if (scs.session->absStartTime() != NULL) {
        rtspClient->sendPlayCommand(*scs.session, continueAfterPLAY,
                scs.session->absStartTime(), scs.session->absEndTime());
    } else {
        scs.duration = scs.session->playEndTime()
                - scs.session->playStartTime();
        rtspClient->sendPlayCommand(*scs.session, continueAfterPLAY);
    }
}

void continueAfterSETUP(RTSPClient* rtspClient, int resultCode,
        char* resultString) {
    do {
        UsageEnvironment& env = rtspClient->envir(); // alias
        StreamClientState& scs = ((ourRTSPClient*) rtspClient)->scs; // alias

        if (resultCode != 0) {
            env << *rtspClient << "Failed to set up the \"" << *scs.subsession
                    << "\" subsession: " << resultString << "\n";
            break;
        }

        env << *rtspClient << "Set up the \"" << *scs.subsession
                << "\" subsession (";
        if (scs.subsession->rtcpIsMuxed()) {
            env << "client port " << scs.subsession->clientPortNum();
        } else {
            env << "client ports " << scs.subsession->clientPortNum() << "-"
                    << scs.subsession->clientPortNum() + 1;
        }
        env << ")\n";

        scs.subsession->sink = HEVCSink::createNew(env, *scs.subsession,
                (void *) rtspClient);
        if (scs.subsession->sink == NULL) {
            env << *rtspClient << "Failed to create a data sink for the \""
                    << *scs.subsession << "\" subsession: "
                    << env.getResultMsg() << "\n";
            break;
        }

        env << *rtspClient << "Created a data sink for the \""
                << *scs.subsession << "\" subsession\n";
        scs.subsession->miscPtr = rtspClient; // a hack to let subsession handle functions get the "RTSPClient" from the subsession
        scs.subsession->sink->startPlaying(*(scs.subsession->readSource()),
                subsessionAfterPlaying, scs.subsession);
        if (scs.subsession->rtcpInstance() != NULL) {
            scs.subsession->rtcpInstance()->setByeHandler(subsessionByeHandler,
                    scs.subsession);
        }
    } while (0);
    delete[] resultString;

    setupNextSubsession(rtspClient);
}

void continueAfterPLAY(RTSPClient* rtspClient, int resultCode,
        char* resultString) {
    Boolean success = False;
    do {
        UsageEnvironment& env = rtspClient->envir(); // alias
        StreamClientState& scs = ((ourRTSPClient*) rtspClient)->scs; // alias

        if (resultCode != 0) {
            env << *rtspClient << "Failed to start playing session: "
                    << resultString << "\n";
            __android_log_print(ANDROID_LOG_INFO, "native",
                    "Failed to start playing session:%s", resultString);
            break;
        }

        if (scs.duration > 0) {
            unsigned const delaySlop = 2; // number of seconds extra to delay, after the stream's expected duration.  (This is optional.)
            scs.duration += delaySlop;
            unsigned uSecsToDelay = (unsigned) (scs.duration * 1000000);
            scs.streamTimerTask = env.taskScheduler().scheduleDelayedTask(
                    uSecsToDelay, (TaskFunc*) streamTimerHandler, rtspClient);
        }

        success = True;
    } while (0);
    delete[] resultString;

    if (!success) {
        __android_log_print(ANDROID_LOG_INFO, "native", "continueAfterPLAY shutdownStream");
        shutdownStream(rtspClient);
    }
}

void subsessionAfterPlaying(void* clientData) {
    MediaSubsession* subsession = (MediaSubsession*) clientData;
    RTSPClient* rtspClient = (RTSPClient*) (subsession->miscPtr);

    Medium::close(subsession->sink);
    subsession->sink = NULL;
    MediaSession& session = subsession->parentSession();
    MediaSubsessionIterator iter(session);
    while ((subsession = iter.next()) != NULL) {
        if (subsession->sink != NULL)
            return; // this subsession is still active
    }

    shutdownStream(rtspClient);
}

void subsessionByeHandler(void* clientData) {
    MediaSubsession* subsession = (MediaSubsession*) clientData;
    RTSPClient* rtspClient = (RTSPClient*) subsession->miscPtr;
    UsageEnvironment& env = rtspClient->envir(); // alias
    CustomClient *client = ((ourRTSPClient *) rtspClient)->mCustomClient;
    client->mRtcpSendBye = true;
    subsessionAfterPlaying(subsession);
}

void streamTimerHandler(void* clientData) {
    ourRTSPClient* rtspClient = (ourRTSPClient*) clientData;
    StreamClientState& scs = rtspClient->scs; // alias
    scs.streamTimerTask = NULL;
    shutdownStream(rtspClient);
}

void shutdownStream(RTSPClient * _client, int exitCode) {
    CustomClient * custom_client = ((ourRTSPClient *) _client)->mCustomClient;

    if ((custom_client->mThreadId != 0)) {
        custom_client->mQuit = 1;
    }
}

// Implementation of "ourRTSPClient":

ourRTSPClient* ourRTSPClient::createNew(UsageEnvironment& env,
        void * client_data, int verbosityLevel, char const* applicationName,
        portNumBits tunnelOverHTTPPortNum) {
    return new ourRTSPClient(env, client_data, verbosityLevel, applicationName,
            tunnelOverHTTPPortNum);
}

ourRTSPClient::ourRTSPClient(UsageEnvironment& env, void * client_data,
        int verbosityLevel, char const* applicationName,
        portNumBits tunnelOverHTTPPortNum) :
        RTSPClient(env, ((CustomClient*) client_data)->mUrl, verbosityLevel,
                applicationName, tunnelOverHTTPPortNum, -1) {
    mCustomClient = (CustomClient*) client_data;
}

ourRTSPClient::~ourRTSPClient() {
}

StreamClientState::StreamClientState() :
        iter(NULL), session(NULL), subsession(NULL), streamTimerTask(NULL), duration(
                0.0) {
}

StreamClientState::~StreamClientState() {
    delete iter;
    if (session != NULL) {
        UsageEnvironment& env = session->envir(); // alias
        env.taskScheduler().unscheduleDelayedTask(streamTimerTask);
        Medium::close(session);
        session = NULL;
    }
}

#define HEVC_SINK_RECEIVE_BUFFER_SIZE 500000

HEVCSink* HEVCSink::createNew(UsageEnvironment& env,
        MediaSubsession& subsession, void * client_data) {
    return new HEVCSink(env, subsession, client_data);
}

HEVCSink::HEVCSink(UsageEnvironment& env, MediaSubsession& subsession,
        void * client_data) :
        MediaSink(env), fHaveWrittenFirstFrame(False), fSubsession(subsession) {
    mClient = (RTSPClient *) client_data;
    CustomClient * client = ((ourRTSPClient *) client_data)->mCustomClient;
    fPrevPresentationTime.tv_sec = ~0;
    fPrevPresentationTime.tv_usec = 0;
    fBuffer = new unsigned char[HEVC_SINK_RECEIVE_BUFFER_SIZE];
    fStreamId = strDup(client->mUrl);
#ifdef FILE_SAVE_ENB
    fd = open("/sdcard/jni-recv.264",O_CREAT | O_WRONLY | O_TRUNC,655);
#endif /* FILE_SAVE_ENB */
#ifdef AUDIO_FILE_SAVE_ENB
    audio_fd = open("/sdcard/jni-recv.aac",O_CREAT | O_WRONLY | O_TRUNC,655);
#endif /* AUDIO_FILE_SAVE_ENB */
    if (strcmp(subsession.codecName(), "H264") == 0) {
        fSPropParameterSetsStr[0] = subsession.fmtp_spropparametersets();
        fSPropParameterSetsStr[1] = NULL;
        fSPropParameterSetsStr[2] = NULL;
#ifdef CHECK_IDRFRM_ENB
        snprintf(codecName, 16, "%s", "H264");
#endif /* CHECK_IDRFRM_ENB */
    } else if (strcmp(subsession.codecName(), "H265") == 0) {
        fSPropParameterSetsStr[0] = fSubsession.fmtp_spropvps();
        fSPropParameterSetsStr[1] = fSubsession.fmtp_spropsps();
        fSPropParameterSetsStr[2] = fSubsession.fmtp_sproppps();
#ifdef CHECK_IDRFRM_ENB
        snprintf(codecName, 16, "%s", "H265");
#endif /* CHECK_IDRFRM_ENB */
    } else {
#ifdef CHECK_IDRFRM_ENB
        codecName[0] = '\0';
#endif /* CHECK_IDRFRM_ENB */
    }
}

HEVCSink::~HEVCSink() {
#ifdef FILE_SAVE_ENB
    if( (fd > 0) ) {::close(fd);}
#endif /* FILE_SAVE_ENB */ 
#ifdef AUDIO_FILE_SAVE_ENB
    if( (audio_fd > 0) ) {::close(audio_fd);}
#endif /* AUDIO_FILE_SAVE_ENB */
    delete[] fBuffer;
    fBuffer = NULL;
    free(fStreamId);
}

void HEVCSink::afterGettingFrame(void* clientData, unsigned frameSize,
        unsigned numTruncatedBytes, struct timeval presentationTime,
        unsigned /*durationInMicroseconds*/) {
    ((HEVCSink*)clientData)->afterGettingFrame(frameSize, numTruncatedBytes, presentationTime);
}

void HEVCSink::afterGettingFrame(unsigned frameSize, unsigned numTruncatedBytes,
        struct timeval presentationTime) {
    unsigned char const start_code[4] = { 0x00, 0x00, 0x00, 0x01 };
    unsigned int timestamp = presentationTime.tv_sec * 1000 + presentationTime.tv_usec / 1000;
    if (strcmp(fSubsession.mediumName(), "video") != 0) {
        if ((fBuffer != NULL)) {
            JNIEnv *env = NULL;
            env = GetEnv();
            jbyteArray jarrRV = env->NewByteArray(frameSize);
            jbyte *jby = env->GetByteArrayElements(jarrRV, 0);
            memcpy(jby, fBuffer, frameSize);
            env->SetByteArrayRegion(jarrRV, 0, frameSize, jby);
            {
                Autolock _l(&((ourRTSPClient*)mClient)->mCustomClient->mRTMPSinkLock);
                if (((ourRTSPClient*)mClient)->mCustomClient->mRTMPSink) {
                    if (!((ourRTSPClient*)mClient)->mCustomClient->mStarted) {
                        int ret = RTMPSink_Write(((ourRTSPClient*)mClient)->mCustomClient->mRTMPSink, fBuffer, frameSize, RTMP_FLAG_PAYLOAD_AUDIO | RTMP_FLAG_INCLUDE_TIMESTAMP, &timestamp);
                        if (ret) {
                            ALOGEA("RTMPSink_Write failed");
                        }
                    }
                }
            }
            audionotify(((ourRTSPClient*) mClient)->mCustomClient,
                    (int) frameSize, jarrRV,
                    (1000000 * presentationTime.tv_sec
                        + presentationTime.tv_usec) & 0x00000000ffffffff);
#ifdef AUDIO_FILE_SAVE_ENB
            {
                unsigned char *adts_header = (unsigned char *)fBuffer;
                adts_header[0] = 0xFF;
                adts_header[1] = 0xF1;
                adts_header[2] = 0x50;
                adts_header[3] = (0x80|(frameSize&0x18)>>11);
                adts_header[4] = ((frameSize&0x7F8)>>3);
                adts_header[5] = (0x1F|(frameSize&0x7)<<5);
                adts_header[6] = 0xFC;
                //addData(adts_header, sizeof(adts_header));
            }
            addData(fBuffer, frameSize);
#endif /* AUDIO_FILE_SAVE_ENB */
            env->ReleaseByteArrayElements(jarrRV, jby, 0);
            env->DeleteLocalRef(jarrRV);
        }
        continuePlaying();
        return;
    }

#ifdef CHECK_IDRFRM_ENB
    if (!fHaveWrittenFirstFrame) {
        if (!strncmp(codecName, "H264", strlen("H264"))) {
            if ((fBuffer[0] != 0x65)) {
                goto SKIP_FRAME;
            }
        } else if (!strncmp(codecName, "H265", strlen("H265"))) {
            if ((fBuffer[0] != 0x26)) {
                goto SKIP_FRAME;
            }
        }
    }
#endif /* CHECK_IDRFRM_ENB */
    if ((fBuffer != NULL)) {
        JNIEnv *env = NULL;
        // add vps sps pps
        unsigned sumByte = 0, writeByte = 0;
        unsigned numSPropRecords[3] = { 0, 0, 0 };
        SPropRecord* sPropRecords[3] = { NULL, NULL, NULL };
        env = GetEnv();
        if (!fHaveWrittenFirstFrame) {
            for (unsigned j = 0; j < 3; ++j) {
                sPropRecords[j] = parseSPropParameterSets(
                        fSPropParameterSetsStr[j], numSPropRecords[j]);
                for (unsigned i = 0; i < numSPropRecords[j]; ++i) {
                    sumByte += (4 + sPropRecords[j][i].sPropLength);
                }
            }
        }
        sumByte += (4 + frameSize);
        jbyteArray jarrRV = env->NewByteArray(sumByte);
        jbyte *jby = env->GetByteArrayElements(jarrRV, 0);
        if (!fHaveWrittenFirstFrame) {
            for (unsigned j = 0; j < 3; ++j) {
                for (unsigned i = 0; i < numSPropRecords[j]; ++i) {
                    memcpy(jby + writeByte, start_code, 4);
                    writeByte += 4;
                    memcpy(jby + writeByte, sPropRecords[j][i].sPropBytes,
                        sPropRecords[j][i].sPropLength);
                    writeByte += sPropRecords[j][i].sPropLength;
                    {
                        Autolock _l(&((ourRTSPClient*)mClient)->mCustomClient->mRTMPSinkLock);
                        if (((ourRTSPClient*)mClient)->mCustomClient->mRTMPSink) {
                            int parameter_type = ((sPropRecords[j][i].sPropBytes[0] & 0x1F) == 0x07 ? PARAMETER_VIDEO_H264_SPS : PARAMETER_VIDEO_H264_PPS);
                            int ret = RTMPSink_SetParameter(((ourRTSPClient*)mClient)->mCustomClient->mRTMPSink, parameter_type, sPropRecords[j][i].sPropBytes, sPropRecords[j][i].sPropLength);
                            if (ret) {
                                ALOGE("RTMPSink_SetParameter %d failed", parameter_type);
                            }
                        }
                    }
#ifdef FILE_SAVE_ENB
                    addData(start_code, 4);
                    addData(sPropRecords[j][i].sPropBytes, sPropRecords[j][i].sPropLength);
#endif /* FILE_SAVE_ENB */
                }
                delete[] sPropRecords[j];
            }
            fHaveWrittenFirstFrame = True; // for next time
        }
        memcpy(jby + writeByte, start_code, 4);
#ifdef FILE_SAVE_ENB
        addData(start_code, 4);
#endif /* FILE_SAVE_ENB */
        writeByte += 4;
        memcpy(jby + writeByte, fBuffer, frameSize);
#ifdef FILE_SAVE_ENB
        addData(fBuffer, frameSize);
#endif /* FILE_SAVE_ENB */
        writeByte += frameSize;

        {
            Autolock _l(&((ourRTSPClient*)mClient)->mCustomClient->mRTMPSinkLock);
            if (((ourRTSPClient*)mClient)->mCustomClient->mRTMPSink) {
                if (((ourRTSPClient*)mClient)->mCustomClient->mStarted) {
                    if ((fBuffer[0] & 0x1F) == 0x05) {
                        RTMPSink_SetParameter(((ourRTSPClient*)mClient)->mCustomClient->mRTMPSink, PARAMETER_TIMESTAMP, &timestamp, sizeof(timestamp));
                        ((ourRTSPClient*)mClient)->mCustomClient->mStarted = false;
                    }
                }
                if (!((ourRTSPClient*)mClient)->mCustomClient->mStarted) {
                    int ret = RTMPSink_Write(((ourRTSPClient*)mClient)->mCustomClient->mRTMPSink, fBuffer, frameSize, RTMP_FLAG_SINGLE_NALU | RTMP_FLAG_INCLUDE_TIMESTAMP, &timestamp);
                    if (ret) {
                        ALOGEV("RTMPSink_Write failed");
                    }
                }
            }
        }
        env->SetByteArrayRegion(jarrRV, 0, writeByte, jby);
        videonotify(((ourRTSPClient*) mClient)->mCustomClient, (int) writeByte,
                jarrRV,
                (1000000 * presentationTime.tv_sec + presentationTime.tv_usec)
                        & 0x00000000ffffffff);

        env->ReleaseByteArrayElements(jarrRV, jby, 0);
        env->DeleteLocalRef(jarrRV);
    }

#ifdef CHECK_IDRFRM_ENB
    SKIP_FRAME:
#endif /* CHECK_IDRFRM_ENB */
    continuePlaying();
}

void HEVCSink::addData(unsigned char const* data, unsigned dataSize) {
#ifdef FILE_SAVE_ENB
    write(fd, data, dataSize);
#endif /* FILE_SAVE_ENB */
#ifdef AUDIO_FILE_SAVE_ENB
    write(audio_fd, data, dataSize);
#endif /* AUDIO_FILE_SAVE_ENB */
}

Boolean HEVCSink::continuePlaying() {
    if (fSource == NULL)
        return False;

    fSource->getNextFrame(fBuffer, HEVC_SINK_RECEIVE_BUFFER_SIZE,
            afterGettingFrame, this, onSourceClosure, this);
    return True;
}

static void* SetupClient(void* _client) {
    CustomClient *client = (CustomClient *) _client;
    TaskScheduler* scheduler = BasicTaskScheduler::createNew();
    client->mEnv = BasicUsageEnvironment::createNew(*scheduler);

    do {
        client->mClient = ourRTSPClient::createNew(*(client->mEnv), client, RTSP_CLIENT_VERBOSITY_LEVEL, "RTSPClient");
    } while (client->mClient == NULL);

    client->mClient->sendDescribeCommand(continueAfterDESCRIBE);

    OutPacketBuffer::maxSize = 600000;
    client->mQuit = 0;
    client->mEnv->taskScheduler().doEventLoop(&(client->mQuit));

    StreamClientState& scs = ((ourRTSPClient*)client->mClient)->scs;

    if (scs.session != NULL) {
        MediaSubsessionIterator iter(*scs.session);
        MediaSubsession* subsession;

        while ((subsession = iter.next()) != NULL) {
            if (subsession->sink != NULL) {
                Medium::close(subsession->sink);
                subsession->sink = NULL;
                if (subsession->rtcpInstance() != NULL) {
                    subsession->rtcpInstance()->setByeHandler(NULL, NULL); // in case the server sends a RTCP "BYE" while handling "TEARDOWN"
                }
            }
        }
    }

    Medium::close(client->mClient);

    if ((NULL != scheduler)) {
        delete scheduler;
        scheduler = NULL;
    }

    if ((NULL != client->mEnv)) {
        client->mEnv->reclaim();
    }

    exitnotify(client);

    DetachCurrent();
    return NULL;
}

//=======================native function==============
//static const char* const kClass_RenderActivity = "com/ipc/mediacodec/VideoDecoder";
static const char* const kClass_RenderActivity =
        "limedia/sdk/api/realtime/TSRTSPNative";
//static const char* const kClass_RenderActivity ="com/android/mediacodec/VideoDecoder";

static void videonotify(CustomClient * client, int msg, const jbyteArray cbb,
        long presentationTime) {
    JNIEnv *env = GetEnv();
    long long newpresentationTime = presentationTime;
    if (cbb != NULL) {
        if (gObject)
            env->CallVoidMethod(gObject, client->mVideoEvent, msg, cbb,
                newpresentationTime);
    } else {
        if (gObject)
            env->CallVoidMethod(gObject, client->mVideoEvent, msg, NULL);
    }
    if (env->ExceptionCheck()) {
        env->ExceptionClear();
    }
}

static void audionotify(CustomClient * client, int msg, const jbyteArray cbb,
        long presentationTime) {
    JNIEnv *env = GetEnv();
    long long newpresentationTime = presentationTime;
    if (cbb != NULL) {
        if (gObject)
            env->CallVoidMethod(gObject, client->mAudioEvent, msg, cbb,
                newpresentationTime);
    } else {
        if (gObject)
            env->CallVoidMethod(gObject, client->mAudioEvent, msg, NULL);
    }
    if (env->ExceptionCheck()) {
        env->ExceptionClear();
    }
}

static void exitnotify(CustomClient *client) {
    JNIEnv *env = GetEnv();
    if (gObject)
        env->CallVoidMethod(gObject, client->mExitEvent);
    if (env->ExceptionCheck()) {
        env->ExceptionClear();
    }
}

static void* thread_func_rtmp_init(void *_url) {
	FUNC_ENTER();
	const char * url = (const char *)_url;

	if (!url) {
		ALOGE("rtmp url is null");
		goto END;
	}

	{
		Autolock _l(&gRTMPSinkLock);
		if (gRTMPSink) {
			/* FIXME: we need to reconnect to another url */
			Autolock _l(&gClients[gRTMPChannel].mRTMPSinkLock);
			gClients[gRTMPChannel].mRTMPSink = NULL;
			goto END;
		}

		gRTMPSink = RTMPSink_Create(url, RTMP_VIDEO_CODEC, RTMP_AUDIO_CODEC);
		if (!gRTMPSink) {
			ALOGE("create RTMPSink failed.");
			goto END;
		}

		int ret = RTMPSink_SetParameter(gRTMPSink, PARAMETER_VIDEO_H264_STARTCODE, (char[]){0,0,0,1}, 4);
		if (ret) {
			ALOGE("set parameter of h264 start code failed.");
			goto END;
		}

		unsigned char asc[2] = {0x12, 0x10};
		ret = RTMPSink_SetParameter(gRTMPSink, PARAMETER_AUDIO_AAC_AUDIOSPECIFICCONFIG, asc, sizeof(asc));
		if (ret) {
			ALOGE("set parameter of aac audio specific config failed.");
			goto END;
		}

		{
			Autolock _l(&gClients[gRTMPChannel].mRTMPSinkLock);
			gClients[gRTMPChannel].mRTMPSink = gRTMPSink;
			gClients[gRTMPChannel].mStarted = true;
		}
	}

END:
	free((void *)url);
	return NULL;
}

static int rtmp_init(const char *url)
{
	FUNC_ENTER();
	int ret;

	pthread_t tid;
	ret = pthread_create(&tid, NULL, thread_func_rtmp_init, (void *)(url?strdup(url):NULL));
	if (ret != 0) {
		ALOGE("create rtmp init thread failed.");
		goto END;
	}

	ret = pthread_detach(tid);
END:
	FUNC_EXIT();
	return ret;
}

void native_init(JNIEnv *env, jobject thiz,jstring rtmp_url, jstring /*rtmp_url*/, jstring channel0, jstring channel1, jstring channel2) {
    ALOGD("RTMP version : %d", RTMPSink_GetVersion());
    {
        if (rtmp_url) {
            const char *jni_url = env->GetStringUTFChars(rtmp_url, 0);
            ALOGD("RTMP url : [%s]", jni_url);
            if (jni_url != NULL && *jni_url) {
                rtmp_init(jni_url);
            }
            env->ReleaseStringUTFChars(rtmp_url, jni_url);
        }
    }

    static bool inited = false;
    if (!inited) {
        gClass = env->GetObjectClass(thiz);
        gClients[0].mVideoEvent = env->GetMethodID(gClass,"postVideoDataFromNative0", "(I[BJ)V");
        gClients[0].mAudioEvent = env->GetMethodID(gClass,"postAudioDataFromNative0", "(I[BJ)V");
        gClients[0].mExitEvent = env->GetMethodID(gClass, "exitNotify0", "()V");
        gClients[1].mVideoEvent = env->GetMethodID(gClass,"postVideoDataFromNative1", "(I[BJ)V");
        gClients[1].mAudioEvent = env->GetMethodID(gClass,"postAudioDataFromNative1", "(I[BJ)V");
        gClients[1].mExitEvent = env->GetMethodID(gClass, "exitNotify1", "()V");
        gClients[2].mVideoEvent = env->GetMethodID(gClass,"postVideoDataFromNative2", "(I[BJ)V");
        gClients[2].mAudioEvent = env->GetMethodID(gClass,"postAudioDataFromNative2", "(I[BJ)V");
        gClients[2].mExitEvent = env->GetMethodID(gClass, "exitNotify2", "()V");
        inited = true;
    }
    {
        /* check if no channel started */
        LockGlobalObject();
        if (gObject == NULL) {
            int i=0;
            for (; i < 3; i++) {
                if (gClients[i].mUrl != NULL)
                    break;
            }
            if (i == 3) {
                gObject = env->NewGlobalRef(thiz);
            }
        }
        UnlockGlobalObject();
    }

    jstring* channels[3] = {
        (channel0 != NULL ? &channel0 : NULL),
        (channel1 != NULL ? &channel1 : NULL),
        (channel2 != NULL ? &channel2 : NULL)
    };
    for (int i = 0; i < 3; i++) {
        CustomClient *client = gClients + i;
        if (channels[i] != NULL) {
            if (client->mUrl == NULL) {
                client->mUrl = env->GetStringUTFChars(*(channels[i]), 0);
                if (client->mUrl != NULL) {
                    char * url = strdup(client->mUrl);
                    env->ReleaseStringUTFChars(*(channels[i]),client->mUrl);
                    client->mUrl = url;
                    if ((pthread_create(&(client->mThreadId), NULL, SetupClient, (void*) client) < 0)) {
                        ALOGD("create thread failed");
                    }
                }
            } else {
                ALOGE("channel %d already in use", i);
            }
        }
    }
}

void native_deinit(JNIEnv *env, jobject thiz, jint mask) {
    for (int i = 0; i < 3; i++) {
        if (mask & (1 << i)) {
            CustomClient * client = gClients + i;
            if (!client->mRtcpSendBye) {
                UsageEnvironment& env = *(client->mEnv); // alias
                StreamClientState& scs = ((ourRTSPClient*) (client->mClient))->scs; // alias
                if (client->mClient && scs.session != NULL) {
                    ALOGD("send teardown command");
                    client->mClient->sendTeardownCommand(*scs.session, NULL);
                }
            }

            client->mRtcpSendBye = false; // false is important

            client->mQuit = 1;
            pthread_join(client->mThreadId, NULL);
            {
                /* clear fields */
                free((void *)(client->mUrl));
                client->mUrl = NULL;
                client->mClient = NULL;
                client->mRTMPSink = NULL;
                client->mStarted = false;
            }
            {
                /* check if all channel released */
                LockGlobalObject();
                if (gObject != NULL) {
                    int i = 0;
                    for (; i < 3; i++) {
                        if (gClients[i].mUrl != NULL)
                            break;
                    }
                    if (i == 3) {
                        env->DeleteGlobalRef(gObject);
                        gObject = NULL;
                        /* destroy RTMPSink if neccessary */
                        Autolock _l(&gRTMPSinkLock);
                        RTMPSink_Destroy(&gRTMPSink);
                        gRTMPChannel = 0;
                    }
                }
                UnlockGlobalObject();
            }
        }
    }
}

void native_forward(JNIEnv *env, jobject thiz, jint channel) {
	if (channel < 1 || channel > 3) {
		return;
	}
	--channel;

	if (gRTMPChannel >=0 && gRTMPChannel <=2) {
		Autolock _l(&gClients[gRTMPChannel].mRTMPSinkLock);
		gClients[gRTMPChannel].mRTMPSink = NULL;
		gClients[gRTMPChannel].mStarted = false;
	}

	Autolock _l(&gClients[channel].mRTMPSinkLock);
	gClients[channel].mRTMPSink = gRTMPSink;
	gClients[channel].mStarted = true;
	gRTMPChannel = channel;

	ALOGD("set channel %d", gRTMPChannel);
}

static JNINativeMethod gMethods[] = {
	{
		"native_init",
		"(Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;Ljava/lang/String;)V",
		(void *) native_init
	},
	{
		"native_deinit",
		"(I)V",
		(void *) native_deinit
	},
	{
		"native_forward",
		"(I)V",
		(void *) native_forward
	}
};

static int register_jni_methods(JNIEnv *env) {
	jclass clazz = (env)->FindClass(kClass_RenderActivity);
	return (env)->RegisterNatives(clazz, gMethods, 3);
}

jint JNI_OnLoad(JavaVM* vm, void* /* reserved */) {
	g_JavaVM = vm;
	JNIEnv* env = NULL;
	jint result = -1;

	if (vm->GetEnv((void**) &env, JNI_VERSION_1_4) != JNI_OK || env == NULL) {
		goto bail;
	}

	if (register_jni_methods(env) < 0) {
		goto bail;
	}

	/* success -- return valid version number */
	result = JNI_VERSION_1_4;

bail:
	return result;
}

static JNIEnv *GetEnv() {
	JNIEnv *env = NULL;

	if (g_JavaVM->GetEnv((void **) &env, JNI_VERSION_1_4) >= 0) {
		goto END;
	}

	if (g_JavaVM->AttachCurrentThread(&env, NULL) < 0) {
		env = NULL;
	}

END:
	return env;
}

static inline void DetachCurrent() {
	g_JavaVM->DetachCurrentThread();
}
